package org.korosoft.rusalad.syncnrun.tools;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.concurrent.atomic.AtomicLong;

/**
 * {@link java.io.InputStream} wrapper counting read data
 *
 * @author Dmitry Korotkov
 * @since 1.0
 */
public class CounterInputStream extends InputStream {
    private static final Log log = LogFactory.getLog(CounterInputStream.class);

    private final InputStream wrapped;
    private final AtomicLong readBytes = new AtomicLong();

    public CounterInputStream(InputStream wrapped) {
        this.wrapped = wrapped;
    }

    public int read() throws IOException {
        int read = wrapped.read();
        if (read >= 0) {
            readBytes.incrementAndGet();
        }
        return read;
    }

    public int read(byte[] b) throws IOException {
        int read = wrapped.read(b);
        if (read >= 0) {
            readBytes.addAndGet(read);
        }
        return read;
    }

    public int read(byte[] b, int off, int len) throws IOException {
        int read = wrapped.read(b, off, len);
        if (read >= 0) {
            readBytes.addAndGet(read);
        }
        return read;
    }

    public long skip(long n) throws IOException {
        long skip = wrapped.skip(n);
        if (skip >= 0) {
            readBytes.addAndGet(skip);
        }
        return skip;
    }

    public int available() throws IOException {
        return wrapped.available();
    }

    public void close() throws IOException {
        wrapped.close();
        if (log.isInfoEnabled()) {
            log.info(String.format("Received %d bytes over the network", readBytes.get()));
        }
    }

    public void mark(int readlimit) {
        wrapped.mark(readlimit);
    }

    public void reset() throws IOException {
        wrapped.reset();
    }

    public boolean markSupported() {
        return wrapped.markSupported();
    }
}
